package com.pw.study.flink.chapter8;

import com.pw.study.flink.entities.OrderEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

public class ExerciseOder {


    public static void main(String[] args) {
        exercise();
    }

    private static void exercise() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        //输入数据
        SingleOutputStreamOperator<OrderEvent> stream = env.readTextFile("data/file/OrderLog.csv").map(ads -> {
            String[] line = ads.split(",");
            return new OrderEvent(Long.valueOf(line[0]), line[1], line[2], Long.valueOf(line[3]) * 1000);
        });

        WatermarkStrategy<OrderEvent> strategy = WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((item, ts) -> item.getEventTime());

        stream.assignTimestampsAndWatermarks(strategy).keyBy(OrderEvent::getOrderId)
                .window(EventTimeSessionWindows.withGap(Time.minutes(45L)))
                .process(new ProcessWindowFunction<OrderEvent, String, Long, TimeWindow>() {

                    private ValueState<OrderEvent> createState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        createState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("createState", OrderEvent.class));
                    }


                    @Override
                    public void process(Long key, Context ctx, Iterable<OrderEvent> elements, Collector<String> out) throws Exception {
                        List<OrderEvent> list = StreamSupport.stream(elements.spliterator(), true).collect(Collectors.toList());
                        String msg = " ";
                        if (list.size() == 2) {
                            msg = "订单key: " + key + " 正常支付。。。";
                        } else {
                            //获取事件
                            OrderEvent ct = list.get(0);

                            if (ct.getEventType().equals("create")) {
                                msg = "更新状态。。。。。。。。。。。。。。。。。";
                                //更新数据
                                createState.update(ct);
                            } else {
                                //从状态中取数据
                                if (createState.value() != null) {
                                    msg = "订单key: " + key + " 超时支付, 请检查系统漏洞。。。";
                                } else {
                                    msg = "订单key： " + key + " 只有支付没有，创建订单。。。。。";

                                }
                            }
                        }

                        out.collect(msg);
                    }
                }).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
